package com.example.actordemo1.io;

import akka.actor.AbstractActor;
import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.io.Tcp;
import akka.io.TcpMessage;
import akka.util.ByteString;

import java.net.InetSocketAddress;


/**
 *   TODO  有问题
 */
public class IoTcp {

   static class Client extends AbstractActor {

       final InetSocketAddress remote;
       final ActorRef listener;

       public static Props props(InetSocketAddress remote, ActorRef listener) {
           return Props.create(Client.class, remote, listener);
       }

       public Client(InetSocketAddress remote, ActorRef listener) {
           this.remote = remote;
           this.listener = listener;

           final ActorRef tcp = Tcp.get(getContext().getSystem()).manager();
           tcp.tell(TcpMessage.connect(remote), getSelf());
       }

       @Override
       public Receive createReceive() {
           return receiveBuilder()
                   .match(
                           Tcp.CommandFailed.class,
                           msg -> {
                               listener.tell("failed", getSelf());
                               getContext().stop(getSelf());
                           })
                   .match(
                           Tcp.Connected.class,
                           msg -> {
                               listener.tell(msg, getSelf());
                               getSender().tell(TcpMessage.register(getSelf()), getSelf());
                               getContext().become(connected(getSender()));
                           })
                   .build();
       }

       private Receive connected(final ActorRef connection) {
           return receiveBuilder()
                   .match(
                           ByteString.class,
                           msg -> {
                               connection.tell(TcpMessage.write((ByteString) msg), getSelf());
                           })
                   .match(
                           Tcp.CommandFailed.class,
                           msg -> {
                               // OS kernel socket buffer was full
                           })
                   .match(
                           Tcp.Received.class,
                           msg -> {
                               listener.tell(msg.data(), getSelf());
                           })
                   .matchEquals(
                           "close",
                           msg -> {
                               connection.tell(TcpMessage.close(), getSelf());
                           })
                   .match(
                           Tcp.ConnectionClosed.class,
                           msg -> {
                               getContext().stop(getSelf());
                           })
                   .build();
       }
   }

    static class Server extends AbstractActor {

        final ActorRef manager;

        public Server(ActorRef manager) {
            this.manager = manager;
        }

        public static Props props(ActorRef manager) {
            return Props.create(Server.class, manager);
        }

        @Override
        public void preStart() throws Exception {
            final ActorRef tcp = Tcp.get(getContext().getSystem()).manager();
            tcp.tell(TcpMessage.bind(getSelf(), new InetSocketAddress("localhost", 1234), 100), getSelf());
        }

        @Override
        public Receive createReceive() {
            return receiveBuilder()
                    .match(
                            Tcp.Bound.class,
                            msg -> {
                                manager.tell(msg, getSelf());
                            })
                    .match(
                            Tcp.CommandFailed.class,
                            msg -> {
                                getContext().stop(getSelf());
                            })
                    .match(
                            Tcp.Connected.class,
                            conn -> {
                                manager.tell(conn, getSelf());
                                final ActorRef handler =
                                        getContext().actorOf(Props.create(SimplisticHandler.class));
                                getSender().tell(TcpMessage.register(handler), getSelf());
                            })
                    .build();
        }
    }

    static class SimplisticHandler extends AbstractActor {
        @Override
        public Receive createReceive() {
            return receiveBuilder()
                    .match(
                            Tcp.Received.class,
                            msg -> {
                                final ByteString data = msg.data();
                                System.out.println(data);
                                getSender().tell(TcpMessage.write(data), getSelf());
                            })
                    .match(
                            Tcp.ConnectionClosed.class,
                            msg -> {
                                getContext().stop(getSelf());
                            })
                    .build();
        }
    }

    public static void main(String[] args) {
        ActorSystem actorSystem = ActorSystem.create("sys");
//        ActorRef serve = actorSystem.actorOf(Server.props(ActorRef.noSender()), "serve1");

        ActorRef client = actorSystem.actorOf(Client.props(new InetSocketAddress("127.0.0.1", 1234),
                ActorRef.noSender()), "client1");
//
//        client.tell(ByteString.fromString("hello"), ActorRef.noSender());

    }

}
